use std::{
    collections::HashMap,
    io::{Read as _, Write},
};

use flate2::{Compression, write::GzEncoder};
use iceberg_ext::catalog::rest::{ErrorModel, IcebergErrorResponse};

use super::{MetadataProperties, io::IOErrorExt};

pub(crate) const PROPERTY_METADATA_COMPRESSION_CODEC: &str = "write.metadata.compression-codec";

#[derive(thiserror::Error, Debug)]
#[error("Unsupported compression codec: {0}")]
pub struct UnsupportedCompressionCodec(String);

impl From<UnsupportedCompressionCodec> for IcebergErrorResponse {
    fn from(value: UnsupportedCompressionCodec) -> Self {
        let typ = "UnsupportedCompressionCodec";
        let boxed = Box::new(value);
        let message = boxed.to_string();

        ErrorModel::bad_request(message, typ, Some(boxed)).into()
    }
}

#[derive(Debug, Clone, Copy, Default)]
pub enum CompressionCodec {
    None,
    #[default]
    Gzip,
}

impl CompressionCodec {
    pub async fn compress(self, payload: Vec<u8>) -> Result<Vec<u8>, IOErrorExt> {
        match self {
            CompressionCodec::None => Ok(payload),
            CompressionCodec::Gzip => tokio::task::spawn_blocking(move || {
                let mut compressed_metadata = GzEncoder::new(Vec::new(), Compression::default());
                compressed_metadata
                    .write_all(&payload)
                    .map_err(|e| IOErrorExt::FileCompression(Box::new(e)))?;

                compressed_metadata
                    .finish()
                    .map_err(|e| IOErrorExt::FileCompression(Box::new(e)))
            })
            .await
            .unwrap_or_else(|e| Err(IOErrorExt::FileCompression(Box::new(e)))),
        }
    }

    pub async fn decompress(self, payload: Vec<u8>) -> Result<Vec<u8>, IOErrorExt> {
        match self {
            CompressionCodec::None => Ok(payload),
            CompressionCodec::Gzip => tokio::task::spawn_blocking(move || {
                let mut decompressed_metadata = Vec::new();
                let mut decoder = flate2::read::GzDecoder::new(payload.as_slice());
                decoder
                    .read_to_end(&mut decompressed_metadata)
                    .map_err(|e| IOErrorExt::FileDecompression(Box::new(e)))?;

                Ok(decompressed_metadata)
            })
            .await
            .unwrap_or_else(|e| Err(IOErrorExt::FileDecompression(Box::new(e)))),
        }
    }

    pub fn as_file_extension(self) -> &'static str {
        match self {
            CompressionCodec::None => "",
            CompressionCodec::Gzip => ".gz",
        }
    }

    pub fn try_from_properties(
        properties: &HashMap<String, String>,
    ) -> Result<Self, UnsupportedCompressionCodec> {
        properties
            .get(PROPERTY_METADATA_COMPRESSION_CODEC)
            .map(String::as_str)
            .map_or(Ok(Self::default()), |value| match value {
                "gzip" => Ok(Self::Gzip),
                "none" => Ok(Self::None),
                unknown => Err(UnsupportedCompressionCodec(unknown.into())),
            })
    }

    pub fn try_from_maybe_properties(
        maybe_properties: Option<&HashMap<String, String>>,
    ) -> Result<Self, UnsupportedCompressionCodec> {
        match maybe_properties {
            Some(properties) => Self::try_from_properties(properties),
            None => Ok(Self::default()),
        }
    }

    pub fn try_from_metadata<T: MetadataProperties>(
        metadata: &T,
    ) -> Result<Self, UnsupportedCompressionCodec> {
        Self::try_from_properties(metadata.properties())
    }
}
